package com.lu;

import cn.hutool.core.util.CharsetUtil;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * @program rabbitmq-demo
 * @description:
 * @author: zhanglu
 * @create: 2022-01-20 14:35:00
 */
@Slf4j
public class RabbitmqApiDemo {

    private static final String HOST = "127.0.0.1";
    private static final Integer PORT = 5672;
    private static final String USER_NAME = "guest";
    private static final String PASSWORD = "guest";

    @Test
    public void init() throws Exception  {
        /*
            exchange: 交换器，交换器将消息路由到一个或多个队列
            RoutingKey: 路由键，生产者把消息发送给交换器，根据路由规则，将该消息路由到队列中
            Queue: 消息队列
         */
        //配置连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(HOST);
        factory.setPort(PORT);
        factory.setUsername(USER_NAME);
        factory.setPassword(PASSWORD);
        //创建一个连接
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        /*
            exchange: 交换机名称
            type: 交换机类型
                fanout: 把消息路由到所有与该交换器绑定的队列中（广播）
                direct: 把消息路由到路由键RoutingKey完全匹配的队列中
                topic: 把消息路由到路由键RoutingKey相匹配的队列中(类似于正则表达式)
                    - `.`分割字符串 `*`模糊匹配一个单词 `#`模糊匹配多个单词(可以是零个)
                    - com.test.lu 可以匹配 .*test.*
                    - com.test.lu.lo 可以匹配 .*test.#
                headers: 消息不依赖于路由键的匹配规则，而是根据发送的消息内容中的headers属性进行匹配。
            durable: 是否持久化。持久化可以将交换机存盘，在服务器重启后不会丢失相关信息
            autoDelete: 是否自动删除。在至少有一个队列或者交换机与该申明的交换机绑定的前提下，之后所有与这个交换机绑定的队列或者交换器在完成使用此交换器后都与此解绑。
            internal: 是否是内置。客户端程序无法直接发送消息到这个交换器中，只能通过交换器路由器到交换器的方式
            argument: 透传的参数
         */
        channel.exchangeDeclare("exchanges.test", BuiltinExchangeType.DIRECT, false, false, null);
        /*
            queue: 队列名称
            exclusive: 是否排他。该队列只对第一个声明他的连接可见，连接关闭后队列自动删除。
                - 排他队列是基于连接可见的。同一个连接的不同信道channel是可以同时访问同一连接创建的排他队列
                - 如果一个连接已经声明了一个排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同
                - 即使该队列是持久化的，一旦连接关闭或者客户端退出胡，该排他队列都会被自动删除
                - 这种队列适用于一个客户端同时发送和读取消息的应用场景。
         */
        channel.queueDeclare("exchanges.test.queue", false, false, false, null);
        channel.queueBind("exchanges.test.queue", "exchanges.test", "exchanges.test.bind");
    }

    @Test
    public void producer() throws Exception {
        //配置连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(HOST);
        factory.setPort(PORT);
        factory.setUsername(USER_NAME);
        factory.setPassword(PASSWORD);
        //创建一个连接
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        String msg = "hello mq";
        channel.basicPublish("exchanges.test", "exchanges.test.bind", null, msg.getBytes(CharsetUtil.UTF_8));
        channel.close();
        connection.close();
    }

    /**
     * rabbitmq 服务端 主动推送消息 至 客户端消费 - 自动提交ack push
     * @throws Exception
     */
    @Test
    public void consumer01() throws Exception {
        //配置连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(HOST);
        factory.setPort(PORT);
        factory.setUsername(USER_NAME);
        factory.setPassword(PASSWORD);
        //创建一个连接
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //自动提交确认ack消息
        channel.basicConsume("exchanges.test.queue", true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] bytes) throws IOException {
                String body = new String(bytes, CharsetUtil.UTF_8);
                log.info(body);
            }
        });
        Thread.currentThread().join();
    }

    /**
     * rabbitmq 服务端 主动推送消息 至 客户端消费 - 手动提交(防止因为程序异常或者客户端意外宕机，消息没有被正确处理，导致漏消息的情况) push
     * @throws Exception
     */
    @Test
    public void consumer02() throws Exception {
        //配置连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(HOST);
        factory.setPort(PORT);
        factory.setUsername(USER_NAME);
        factory.setPassword(PASSWORD);
        //创建一个连接
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //手动提交确认ack消息
        channel.basicConsume("exchanges.test.queue", false, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] bytes) throws IOException {
//                int i = 1/0;
                String body = new String(bytes, CharsetUtil.UTF_8);
                log.info(body);
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
        Thread.currentThread().join();
    }

    /**
     * 客户端消费者主动拉取消息 pull
     * @throws Exception
     */
    @Test
    public void consumer03() throws Exception {
        //配置连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(HOST);
        factory.setPort(PORT);
        factory.setUsername(USER_NAME);
        factory.setPassword(PASSWORD);
        //创建一个连接
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //手动提交确认ack消息
        while (true){
            GetResponse response = channel.basicGet("exchanges.test.queue", false);
            if(response == null){
                log.info("wait 2s");
                TimeUnit.SECONDS.sleep(2);
                continue;
            }
            String body = new String(response.getBody(), CharsetUtil.UTF_8);
            log.info(body);
            channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
            log.info("wait 0.5s");
            TimeUnit.MILLISECONDS.sleep(500);
        }
    }

}
